#!/bin/bash
# 前提：需要安装jdk
flume_version="1.9.0"

#获取本机IP
host_addr=`ifconfig -a|grep inet|grep -v 127.0.0.1|grep -v inet6|awk '{print $2}'|sed -n '$p'`

# 指定KAFKA外网访问地址
read -p "Enter kafka outer visit address,default:$host_addr:9092 " kafkaOuterVisitAddress
if [ -z "$kafkaOuterVisitAddress" ]; then
	kafkaOuterVisitAddress="$host_addr:9092"
fi
echo "your kafka outer visit is: $kafkaOuterVisitAddress"

mkdir -p /data/tools
cd /data/tools

flume_url="https://mirrors.huaweicloud.com/apache/flume/${flume_version}/apache-flume-${flume_version}-bin.tar.gz"

wget -c ${flume_url} || exit 1

tar zxvf apache-flume-${flume_version}-bin.tar.gz || exit 1

ln -s apache-flume-${flume_version}-bin flume || exit 1

# must install jdk first and set jdk environment for flume
cat >/data/tools/flume/flume-env.sh <<EOF
export JAVA_OPTS="-Xms100m -Xmx2048m -Dcom.sun.management.jmxremote"
EOF
# create template configure , modify it According to your environment
cat >/data/tools/flume/conf/tail_to_kafka.conf <<EOF
a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.channels = c1
a1.sources.r1.command = tail -f /data/tools/nginx/log/access.log
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = nginxlog
a1.sinks.k1.brokerList = ###kafkaOuterVisitAddress###
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20
a1.sinks.k1.channel = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
EOF


sed -i 's,###kafkaOuterVisitAddress###,'$kafkaOuterVisitAddress',g' /data/tools/flume/conf/tail_to_kafka.conf

#create startup script
cat >/data/tools/flume/start_tail_to_kafka.sh <<EOF
#!/bin/bash
cd /data/tools/flume
bin/flume-ng agent -c conf/ -f conf/tail_to_kafka.conf -n a1 -Dflume.root.logger=INFO,console   -Dflume.monitoring.type=http -Dflume.monitoring.port=34545 >> /tmp/flume_tail_to_kafka.log 2>&1 &
EOF

chmod +x /data/tools/flume/start_tail_to_kafka.sh
